package org.example.watermarks;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

/**
自定义水印生成器
 **/
public class CustomPeriodicWatermarkExample<T> implements WatermarkGenerator<T> {

    private long maxTimestamp;

    /** The maximum out-of-orderness that this watermark generator assumes. */
    private final long outOfOrdernessMillis;

    public CustomPeriodicWatermarkExample(long outOfOrdernessMillis) {
        this.outOfOrdernessMillis = outOfOrdernessMillis;
        this.maxTimestamp = Long.MIN_VALUE + this.outOfOrdernessMillis +1;
    }

    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
//        System.out.println("断点式：" + new Watermark(maxTimestamp - outOfOrdernessMillis - 1L).toString());
        maxTimestamp = Math.max(eventTimestamp, maxTimestamp);
//        output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1L));
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
//        System.out.println("周期性：" + new Watermark(maxTimestamp - outOfOrdernessMillis - 1L).toString());
        output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1L));
    }
}
